<!DOCTYPE html>
<html xmlns="http://www.w3.org/1999/xhtml">
  <head>

    <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
    <meta content="Cask Data, Inc." name="author" />
<meta content="Copyright © 2014-2016 Cask Data, Inc." name="copyright" />


    <meta name="git_release" content="6.1.1">
    <meta name="git_hash" content="05fbac36f9f7aadeb44f5728cea35136dbc243e5">
    <meta name="git_timestamp" content="2020-02-09 08:22:47 +0800">
    <title>Real-Time Data Processing with a Flow</title>

    <link rel="stylesheet" href="../_static/cdap-bootstrap.css" type="text/css" />
    <link rel="stylesheet" href="../_static/pygments.css" type="text/css" />
    <link rel="stylesheet" href="../_static/bootstrap-3.3.6/css/bootstrap.min.css" type="text/css" />
    <link rel="stylesheet" href="../_static/bootstrap-3.3.6/css/bootstrap-theme.min.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/bootstrap-sphinx.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/cdap-dynamicscrollspy-4.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/jquery.mCustomScrollbar.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/cdap-jquery.mCustomScrollbar.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/abixTreeList-2.css" type="text/css" />
    <link rel="stylesheet" href="../_static/cdap-bootstrap.css" type="text/css" />

    <script type="text/javascript">
      var DOCUMENTATION_OPTIONS = {
        URL_ROOT:    '',
        VERSION:     '6.1.1',
        COLLAPSE_INDEX: false,
        FILE_SUFFIX: '.html',
        HAS_SOURCE:  false
      };
    </script>
    <script type="text/javascript" src="../_static/jquery.js"></script>
    <script type="text/javascript" src="../_static/underscore.js"></script>
    <script type="text/javascript" src="../_static/doctools.js"></script>
    <script type="text/javascript" src="../_static/language_data.js"></script>

    <link rel="shortcut icon" href="../_static/favicon.ico"/>
    <link rel="index" title="Index" href="../genindex.html" />
    <link rel="search" title="Search" href="../search.html" />
    <link rel="top" title="Cask Data Application Platform 6.1.1 Documentation" href="../index.html" />
    <link rel="up" title="How-To Guides" href="index.html" />
    <link rel="next" title="Ingesting Data into CDAP using Apache Flume" href="cdap-flume-guide.html" />
    <link rel="prev" title="Creating Hydrator Applications using CDAP System Artifacts" href="cdap-etl-guide.html" />
    <!-- block extrahead -->
    <meta charset='utf-8'>
    <meta http-equiv='X-UA-Compatible' content='IE=edge,chrome=1'>
    <meta name='viewport' content='width=device-width, initial-scale=1.0, maximum-scale=1'>
    <meta name="apple-mobile-web-app-capable" content="yes">
    <!-- block extrahead end -->

</head>
<body role="document">

<!-- block navbar -->
<div id="navbar" class="navbar navbar-inverse navbar-default navbar-fixed-top">
    <div class="container-fluid">
      <div class="row">
        <div class="navbar-header">
          <!-- .btn-navbar is used as the toggle for collapsed navbar content -->
          <a class="navbar-brand" href="../table-of-contents/../../index.html">
            <span><img alt="CDAP logo" src="../_static/cdap_logo.svg"/></span>
          </a>

          <button type="button" class="navbar-toggle" data-toggle="collapse" data-target=".nav-collapse">
            <span class="icon-bar"></span>
            <span class="icon-bar"></span>
            <span class="icon-bar"></span>
          </button>

          <div class="pull-right">
            <div class="dropdown version-dropdown">
              <a href="#" class="dropdown-toggle" data-toggle="dropdown"
                role="button" aria-haspopup="true" aria-expanded="false">
                v 6.1.1 <span class="caret"></span>
              </a>
              <ul class="dropdown-menu">
                <li><a href="//docs.cdap.io/cdap/5.1.2/en/index.html">v 5.1.2</a></li>
                <li><a href="//docs.cdap.io/cdap/4.3.4/en/index.html">v 4.3.4</a></li>
              </ul>
            </div>
          </div>
          <form class="navbar-form navbar-right navbar-search" action="../search.html" method="get">
            <div class="form-group">
              <div class="navbar-search-image material-icons"></div>
              <input type="text" name="q" class="form-control" placeholder="  Search" />
            </div>
            <input type="hidden" name="check_keywords" value="yes" />
            <input type="hidden" name="area" value="default" />
          </form>

          <div class="collapse navbar-collapse nav-collapse navbar-right navbar-navigation">
            <ul class="nav navbar-nav"><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link " href="../table-of-contents/../../index.html">简介</a></li><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link current" href="../table-of-contents/../../guides.html">手册</a></li><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link " href="../table-of-contents/../../reference-manual/index.html">参考</a></li><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link " href="../table-of-contents/../../faqs/index.html">帮助</a></li>
            </ul>
          </div>

        </div>
      </div>
    </div>
  </div><!-- block navbar end -->
<!-- block main content -->
<div class="main-container container">
  <div class="row"><div class="col-md-2">
      <div id="sidebar" class="bs-sidenav scrollable-y-outside" role="complementary">
<!-- theme_manual: examples-manual -->
<!-- theme_manual_highlight: guides -->
<!-- sidebar_title_link: ../table-of-contents/../../guides.html -->

  <div role="note" aria-label="manuals links"><h3><a href="../table-of-contents/../../guides.html">Guides</a></h3>

    <ul class="this-page-menu">
      <li class="toctree-l1"><a href="../table-of-contents/../../user-guide/index.html" rel="nofollow">用户手册</a>
      </li>
      <li class="toctree-l1"><a href="../table-of-contents/../../developer-manual/index.html" rel="nofollow">开发手册</a>
      </li>
      <li class="toctree-l1"><a href="../table-of-contents/../../admin-manual/index.html" rel="nofollow">管理手册</a>
      </li>
      <li class="toctree-l1"><a href="../table-of-contents/../../integrations/index.html" rel="nofollow">集成手册</a>
      </li>
      <li class="toctree-l1"><b><a href="../table-of-contents/../../examples-manual/index.html" rel="nofollow">最佳实践</a></b>
      <nav class="pagenav">
      <ul class="current">
<li class="toctree-l1"><a class="reference internal" href="../index.html">Introduction</a></li>
<li class="toctree-l1 current"><a class="reference internal" href="index.html">How-To Guides</a><ul class="current">
<li class="toctree-l2"><a class="reference internal" href="cdap-bi-guide.html">Analyzing CDAP Data from BI Tools</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-cube-guide.html">Data Analysis with OLAP Cube</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-etl-guide.html">Creating Hydrator Applications using CDAP System Artifacts</a></li>
<li class="toctree-l2 current"><a class="current reference internal" href="#">Real-Time Data Processing with a Flow</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-flume-guide.html">Ingesting Data into CDAP using Apache Flume</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-kafka-ingest-guide.html">Consuming Data from Kafka</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-mapreduce-guide.html">Batch Data Processing with CDAP</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-spark-guide.html">Iterative Data Processing with Apache Spark</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-timeseries-guide.html">Storing Timeseries Data</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-twitter-ingest-guide.html">Consuming Twitter Data in Real Time</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-workflow-guide.html">Batch Data Processing with CDAP using Workflow</a></li>
</ul>
</li>
</ul>
</nav>
      </li>
    </ul>
  </div></div>
    </div><div class="col-md-8 content" id="main-content">
    
  <div class="section" id="real-time-data-processing-with-a-flow">
<span id="cdap-flow-guide"></span><h1>Real-Time Data Processing with a Flow<a class="headerlink" href="#real-time-data-processing-with-a-flow" title="Permalink to this headline">🔗</a></h1>
<blockquote class="pull-quote">
<div><strong>Source Code Repository:</strong> Source code (and other resources) for this guide are available at the
<a class="reference external" href="https://github.com/cdap-guides/cdap-flow-guide/tree/develop">CDAP Guides GitHub repository</a>.</div></blockquote>
<p>In this guide you will learn how to process data in real time with the Cask Data Application Platform (CDAP).
You will also learn how easy it is to scale out real-time data processing applications on CDAP.</p>
<div class="section" id="what-you-will-build">
<h2>What You Will Build<a class="headerlink" href="#what-you-will-build" title="Permalink to this headline">🔗</a></h2>
<p>You will build a CDAP application that processes disk usage information across machines in a large company in order
to identify disks that may need to be replaced soon. You will:</p>
<ul class="simple">
<li>Build a <a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/building-blocks/flows-flowlets/flows.html">Flow</a>
to process disk usage data in real time;</li>
<li>Use <a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/building-blocks/datasets/index.html">Datasets</a>
to store the number of slow reads per disk and track slow disks that may need to be replaced soon; and</li>
<li>Build a <a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/building-blocks/services.html">Service</a>
to query via HTTP which slow disks should be replaced.</li>
</ul>
</div>
<div class="section" id="what-you-will-need">
<h2>What You Will Need<a class="headerlink" href="#what-you-will-need" title="Permalink to this headline">🔗</a></h2>
<ul class="simple">
<li><a class="reference external" href="http://www.oracle.com/technetwork/java/javase/downloads/index.html">JDK 7 or 8</a></li>
<li><a class="reference external" href="http://maven.apache.org/download.cgi">Apache Maven 3.1+</a></li>
<li><a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/getting-started/standalone/index.html">CDAP SDK</a></li>
</ul>
</div>
<div class="section" id="lets-build-it">
<h2>Let’s Build It!<a class="headerlink" href="#lets-build-it" title="Permalink to this headline">🔗</a></h2>
<p>The following sections will guide you through implementing a real-time data processing application from scratch.
If you want to deploy and run the application right away, you can clone the sources from this GitHub repository.
In that case, feel free to skip the following two sections and jump directly to the
<a class="reference external" href="#build-and-run-application">Build and Run Application</a> section.</p>
<div class="section" id="application-design">
<h3>Application Design<a class="headerlink" href="#application-design" title="Permalink to this headline">🔗</a></h3>
<p>In CDAP, you process data in real time by implementing a Flow. In this example the Flow consists of two Flowlets.
The <code class="docutils literal notranslate"><span class="pre">Detector</span></code> Flowlet consumes data from a <code class="docutils literal notranslate"><span class="pre">diskReads</span></code> Stream, parses the event, and outputs the disk ID if the event was a slow disk read.
The <code class="docutils literal notranslate"><span class="pre">Tracker</span></code> Flowlet reads the disk ID emitted by the <code class="docutils literal notranslate"><span class="pre">Detector</span></code>, and updates a count of how often that disk has recorded a slow read.
If a disk has recorded too many slow reads, the disk ID is written to a separate dataset that tracks all disks that should soon be replaced.</p>
<a class="reference internal image-reference" href="../_images/app-design2.png"><img alt="../_images/app-design2.png" class="align-center" src="../_images/app-design2.png" style="width: 8in;" /></a>
</div>
<div class="section" id="implementation">
<h3>Implementation<a class="headerlink" href="#implementation" title="Permalink to this headline">🔗</a></h3>
<p>The recommended way to build a CDAP application from scratch is to use a Maven project.
Use the following directory structure (you’ll find contents of the files below):</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="go">./pom.xml</span>
<span class="go">./src/main/java/co/cask/cdap/guides/flow/DiskPerformanceApp.java</span>
<span class="go">./src/main/java/co/cask/cdap/guides/flow/DiskPerformanceFlow.java</span>
<span class="go">./src/main/java/co/cask/cdap/guides/flow/DiskPerformanceHTTPHandler.java</span>
<span class="go">./src/main/java/co/cask/cdap/guides/flow/DetectorFlowlet.java</span>
<span class="go">./src/main/java/co/cask/cdap/guides/flow/TrackerFlowlet.java</span>
</pre></div>
</div>
<p>First create the application, which contains a stream, flow, and datasets.</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">DiskPerformanceApp</span> <span class="kd">extends</span> <span class="n">AbstractApplication</span> <span class="p">{</span>
  <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">APP_NAME</span> <span class="o">=</span> <span class="s">&quot;DiskPerformanceApp&quot;</span><span class="p">;</span>
  <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">STREAM_NAME</span> <span class="o">=</span> <span class="s">&quot;diskReads&quot;</span><span class="p">;</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">configure</span><span class="p">()</span> <span class="p">{</span>
    <span class="n">setName</span><span class="p">(</span><span class="n">APP_NAME</span><span class="p">);</span>
    <span class="n">addStream</span><span class="p">(</span><span class="k">new</span> <span class="n">Stream</span><span class="p">(</span><span class="n">STREAM_NAME</span><span class="p">));</span>
    <span class="n">createDataset</span><span class="p">(</span><span class="s">&quot;slowDiskReads&quot;</span><span class="p">,</span> <span class="n">KeyValueTable</span><span class="p">.</span><span class="na">class</span><span class="p">);</span>
    <span class="n">createDataset</span><span class="p">(</span><span class="s">&quot;slowDisks&quot;</span><span class="p">,</span> <span class="n">KeyValueTable</span><span class="p">.</span><span class="na">class</span><span class="p">);</span>
    <span class="n">addFlow</span><span class="p">(</span><span class="k">new</span> <span class="n">DiskPerformanceFlow</span><span class="p">());</span>
    <span class="n">addService</span><span class="p">(</span><span class="n">DiskPerformanceHTTPHandler</span><span class="p">.</span><span class="na">NAME</span><span class="p">,</span> <span class="k">new</span> <span class="n">DiskPerformanceHTTPHandler</span><span class="p">());</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p>Next, we create a Flow, which is composed of two Flowlets, the <code class="docutils literal notranslate"><span class="pre">Detector</span></code> and the <code class="docutils literal notranslate"><span class="pre">Tracker</span></code>.
The Detector Flowlet parses disk I/O events from the Stream and emits the disk ID if the
operation is slower than a threshold. The Tracker consumes the output of the Detector Flowlet
and performs an analysis to detect a slow disk. Since a Tracker Flowlet performs dataset operations,
it may be slower than a Detector Flowlet that performs all processing in memory. Thus, it’s a good idea
to have multiple Tracker Flowlet instances.</p>
<p>In the Flow specification below, we’ll start with a single
Detector and two Tracker Flowlets.
The parser reads from the stream, and the tracker then reads from the parser.
We will set the number of <code class="docutils literal notranslate"><span class="pre">Tracker</span></code> instances to two.
This means that there will be two separate <code class="docutils literal notranslate"><span class="pre">Trackers</span></code> running, each taking turns reading what the Detector outputs.
You want to do this if a single <code class="docutils literal notranslate"><span class="pre">Detector</span></code> can output more quickly than a single <code class="docutils literal notranslate"><span class="pre">Tracker</span></code> can process.</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">DiskPerformanceFlow</span> <span class="kd">extends</span> <span class="n">AbstractFlow</span> <span class="p">{</span>
  <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">NAME</span> <span class="o">=</span> <span class="s">&quot;DiskPerformanceFlow&quot;</span><span class="p">;</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">configure</span><span class="p">()</span> <span class="p">{</span>
    <span class="n">setName</span><span class="p">(</span><span class="n">NAME</span><span class="p">);</span>
    <span class="n">setDescription</span><span class="p">(</span><span class="s">&quot;Tracks slow disks using I/O ops stats&quot;</span><span class="p">);</span>
    <span class="n">addFlowlet</span><span class="p">(</span><span class="n">DetectorFlowlet</span><span class="p">.</span><span class="na">NAME</span><span class="p">,</span> <span class="k">new</span> <span class="n">DetectorFlowlet</span><span class="p">());</span>
    <span class="c1">// start with 2 instances of the tracker</span>
    <span class="n">addFlowlet</span><span class="p">(</span><span class="n">TrackerFlowlet</span><span class="p">.</span><span class="na">NAME</span><span class="p">,</span> <span class="k">new</span> <span class="n">TrackerFlowlet</span><span class="p">(),</span> <span class="mi">2</span><span class="p">);</span>
    <span class="n">connectStream</span><span class="p">(</span><span class="n">DiskPerformanceApp</span><span class="p">.</span><span class="na">STREAM_NAME</span><span class="p">,</span> <span class="n">DetectorFlowlet</span><span class="p">.</span><span class="na">NAME</span><span class="p">);</span>
    <span class="n">connect</span><span class="p">(</span><span class="n">DetectorFlowlet</span><span class="p">.</span><span class="na">NAME</span><span class="p">,</span> <span class="n">TrackerFlowlet</span><span class="p">.</span><span class="na">NAME</span><span class="p">);</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p>Next we create the <code class="docutils literal notranslate"><span class="pre">Detector</span></code> Flowlet, which reads from the Stream and outputs the disk ID if the event was a slow read.</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">DetectorFlowlet</span> <span class="kd">extends</span> <span class="n">AbstractFlowlet</span> <span class="p">{</span>
  <span class="kd">private</span> <span class="kd">static</span> <span class="kd">final</span> <span class="kt">long</span> <span class="n">SLOW_THRESHOLD</span> <span class="o">=</span> <span class="mi">1000</span><span class="p">;</span>
  <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">NAME</span> <span class="o">=</span> <span class="s">&quot;slowReadDetector&quot;</span><span class="p">;</span>

  <span class="kd">private</span> <span class="n">OutputEmitter</span><span class="o">&lt;</span><span class="n">String</span><span class="o">&gt;</span> <span class="n">out</span><span class="p">;</span>

  <span class="nd">@ProcessInput</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">process</span><span class="p">(</span><span class="n">StreamEvent</span> <span class="n">diskMetrics</span><span class="p">)</span> <span class="p">{</span>
    <span class="n">String</span> <span class="n">event</span> <span class="o">=</span> <span class="n">Charsets</span><span class="p">.</span><span class="na">UTF_8</span><span class="p">.</span><span class="na">decode</span><span class="p">(</span><span class="n">diskMetrics</span><span class="p">.</span><span class="na">getBody</span><span class="p">()).</span><span class="na">toString</span><span class="p">();</span>
    <span class="c1">// events are expected to have the following format:</span>
    <span class="c1">// diskId operationTime (in microseconds)</span>
    <span class="n">String</span><span class="o">[]</span> <span class="n">fields</span> <span class="o">=</span> <span class="n">event</span><span class="p">.</span><span class="na">split</span><span class="p">(</span><span class="s">&quot; &quot;</span><span class="p">,</span> <span class="mi">2</span><span class="p">);</span>
    <span class="n">String</span> <span class="n">diskId</span> <span class="o">=</span> <span class="n">fields</span><span class="o">[</span><span class="mi">0</span><span class="o">]</span><span class="p">;</span>
    <span class="kt">long</span> <span class="n">readTime</span> <span class="o">=</span> <span class="n">Long</span><span class="p">.</span><span class="na">parseLong</span><span class="p">(</span><span class="n">fields</span><span class="o">[</span><span class="mi">1</span><span class="o">]</span><span class="p">);</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">readTime</span> <span class="o">&gt;</span> <span class="n">SLOW_THRESHOLD</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">out</span><span class="p">.</span><span class="na">emit</span><span class="p">(</span><span class="n">diskId</span><span class="p">);</span>
    <span class="p">}</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p>Next we create the <code class="docutils literal notranslate"><span class="pre">Tracker</span></code> Flowlet, which reads the output of the <code class="docutils literal notranslate"><span class="pre">Detector</span></code>
Flowlet, and updates how many times each disk reported a slow read. If a disk records too
many slow reads, the <code class="docutils literal notranslate"><span class="pre">Tracker</span></code> places it in a separate dataset used to track slow disks
that may need to be replaced soon.</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">TrackerFlowlet</span> <span class="kd">extends</span> <span class="n">AbstractFlowlet</span> <span class="p">{</span>
  <span class="c1">// intentionally set very low for illustrative purposes</span>
  <span class="kd">private</span> <span class="kd">static</span> <span class="kd">final</span> <span class="kt">long</span> <span class="n">FLAG_THRESHOLD</span> <span class="o">=</span> <span class="mi">3</span><span class="p">;</span>
  <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">NAME</span> <span class="o">=</span> <span class="s">&quot;slowDiskTracker&quot;</span><span class="p">;</span>

  <span class="nd">@UseDataSet</span><span class="p">(</span><span class="s">&quot;slowDiskReads&quot;</span><span class="p">)</span>
  <span class="kd">private</span> <span class="n">KeyValueTable</span> <span class="n">slowDiskReadsTable</span><span class="p">;</span>

  <span class="nd">@UseDataSet</span><span class="p">(</span><span class="s">&quot;slowDisks&quot;</span><span class="p">)</span>
  <span class="kd">private</span> <span class="n">KeyValueTable</span> <span class="n">slowDisksTable</span><span class="p">;</span>

  <span class="nd">@ProcessInput</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">process</span><span class="p">(</span><span class="n">String</span> <span class="n">diskId</span><span class="p">)</span> <span class="p">{</span>
    <span class="kt">byte</span><span class="o">[]</span> <span class="n">countAsBytes</span> <span class="o">=</span> <span class="n">slowDiskReadsTable</span><span class="p">.</span><span class="na">read</span><span class="p">(</span><span class="n">diskId</span><span class="p">);</span>
    <span class="kt">long</span> <span class="n">slowCount</span> <span class="o">=</span> <span class="n">countAsBytes</span> <span class="o">==</span> <span class="kc">null</span> <span class="o">?</span> <span class="mi">0</span> <span class="p">:</span> <span class="n">Bytes</span><span class="p">.</span><span class="na">toLong</span><span class="p">(</span><span class="n">countAsBytes</span><span class="p">);</span>
    <span class="n">slowCount</span><span class="o">++</span><span class="p">;</span>
    <span class="n">slowDiskReadsTable</span><span class="p">.</span><span class="na">write</span><span class="p">(</span><span class="n">diskId</span><span class="p">,</span> <span class="n">Bytes</span><span class="p">.</span><span class="na">toBytes</span><span class="p">(</span><span class="n">slowCount</span><span class="p">));</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">slowCount</span> <span class="o">==</span> <span class="n">FLAG_THRESHOLD</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">slowDisksTable</span><span class="p">.</span><span class="na">write</span><span class="p">(</span><span class="n">diskId</span><span class="p">,</span> <span class="n">Bytes</span><span class="p">.</span><span class="na">toBytes</span><span class="p">(</span><span class="n">System</span><span class="p">.</span><span class="na">currentTimeMillis</span><span class="p">()));</span>
    <span class="p">}</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p>Finally, we implement a Service that exposes a RESTful API used to display the slow disks that need to be replaced soon:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">DiskPerformanceHTTPHandler</span> <span class="kd">extends</span> <span class="n">AbstractHttpServiceHandler</span> <span class="p">{</span>
  <span class="kd">private</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">SimpleDateFormat</span> <span class="n">DATE_FORMAT</span> <span class="o">=</span> <span class="k">new</span> <span class="n">SimpleDateFormat</span><span class="p">(</span><span class="s">&quot;yyyy-MM-dd HH:mm:ss z&quot;</span><span class="p">);</span>
  <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">NAME</span> <span class="o">=</span> <span class="s">&quot;DiskPerformanceService&quot;</span><span class="p">;</span>

  <span class="nd">@UseDataSet</span><span class="p">(</span><span class="s">&quot;slowDisks&quot;</span><span class="p">)</span>
  <span class="kd">private</span> <span class="n">KeyValueTable</span> <span class="n">slowDisksTable</span><span class="p">;</span>

  <span class="nd">@Path</span><span class="p">(</span><span class="s">&quot;slowdisks&quot;</span><span class="p">)</span>
  <span class="nd">@GET</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">getSlowDisks</span><span class="p">(</span><span class="n">HttpServiceRequest</span> <span class="n">request</span><span class="p">,</span> <span class="n">HttpServiceResponder</span> <span class="n">responder</span><span class="p">)</span> <span class="p">{</span>
    <span class="n">Iterator</span><span class="o">&lt;</span><span class="n">KeyValue</span><span class="o">&lt;</span><span class="kt">byte</span><span class="o">[]</span><span class="p">,</span> <span class="kt">byte</span><span class="o">[]&gt;&gt;</span> <span class="n">slowDisksScan</span> <span class="o">=</span> <span class="n">slowDisksTable</span><span class="p">.</span><span class="na">scan</span><span class="p">(</span><span class="kc">null</span><span class="p">,</span> <span class="kc">null</span><span class="p">);</span>
    <span class="n">Map</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="n">slowDisks</span> <span class="o">=</span> <span class="n">Maps</span><span class="p">.</span><span class="na">newHashMap</span><span class="p">();</span>
    <span class="k">while</span> <span class="p">(</span><span class="n">slowDisksScan</span><span class="p">.</span><span class="na">hasNext</span><span class="p">())</span> <span class="p">{</span>
      <span class="n">KeyValue</span><span class="o">&lt;</span><span class="kt">byte</span><span class="o">[]</span><span class="p">,</span> <span class="kt">byte</span><span class="o">[]&gt;</span> <span class="n">slowDisk</span> <span class="o">=</span> <span class="n">slowDisksScan</span><span class="p">.</span><span class="na">next</span><span class="p">();</span>
      <span class="n">String</span> <span class="n">diskId</span> <span class="o">=</span> <span class="n">Bytes</span><span class="p">.</span><span class="na">toString</span><span class="p">(</span><span class="n">slowDisk</span><span class="p">.</span><span class="na">getKey</span><span class="p">());</span>
      <span class="kt">long</span> <span class="n">troubleTime</span> <span class="o">=</span> <span class="n">Bytes</span><span class="p">.</span><span class="na">toLong</span><span class="p">(</span><span class="n">slowDisk</span><span class="p">.</span><span class="na">getValue</span><span class="p">());</span>
      <span class="n">String</span> <span class="n">troubleTimeStr</span> <span class="o">=</span> <span class="n">DATE_FORMAT</span><span class="p">.</span><span class="na">format</span><span class="p">(</span><span class="k">new</span> <span class="n">Date</span><span class="p">(</span><span class="n">troubleTime</span><span class="p">));</span>
      <span class="n">slowDisks</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="n">diskId</span><span class="p">,</span> <span class="n">troubleTimeStr</span><span class="p">);</span>
    <span class="p">}</span>
    <span class="n">responder</span><span class="p">.</span><span class="na">sendJson</span><span class="p">(</span><span class="mi">200</span><span class="p">,</span> <span class="n">slowDisks</span><span class="p">);</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p>With this, we have a working application!
We can build it, send data to the stream, and send an HTTP request to get slow disks that should be replaced soon.
Before we do that, let’s add a couple enhancements.</p>
</div>
<div class="section" id="real-time-processing-with-micro-batches">
<h3>Real-Time Processing with Micro-batches<a class="headerlink" href="#real-time-processing-with-micro-batches" title="Permalink to this headline">🔗</a></h3>
<p>Everything that happens in the process method of a flowlet is guaranteed to happen exactly once.
This is made possible by the execution of each process method inside a separate transaction, which is done by the CDAP framework.
The overhead of the transaction is very small, but it is a good idea to minimize it even further by instructing the framework to
process multiple inputs within the same transaction. That is, consume up to a small number of inputs, if those are available.
This technique is called “processing with micro-batches.”</p>
<p>With a batch size of 100, we will pay the cost of the overhead just once for every 100 events instead of 100 times for 100 events.
Telling a flowlet to process its input in batches of 100 is as simple as adding the Batch annotation to the process method.</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">TrackerFlowlet</span> <span class="kd">extends</span> <span class="n">AbstractFlowlet</span> <span class="p">{</span>
  <span class="p">...</span>

  <span class="nd">@ProcessInput</span>
  <span class="nd">@Batch</span><span class="p">(</span><span class="mi">100</span><span class="p">)</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">process</span><span class="p">(</span><span class="n">String</span> <span class="n">diskId</span><span class="p">)</span> <span class="p">{</span>
    <span class="p">...</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p>Most of the time, using mini-batches is a trade-off between processing latency and throughput.
You pay less overhead for transactions with mini-batches in order to reach a higher throughput.
At the same time, your event is processed only when the whole batch is processed, which usually means a higher latency.</p>
</div>
<div class="section" id="optimizing-for-scale">
<h3>Optimizing for Scale<a class="headerlink" href="#optimizing-for-scale" title="Permalink to this headline">🔗</a></h3>
<p>When using micro-batching, you need to be also careful about two things:</p>
<ul class="simple">
<li>running out of memory; and</li>
<li>the increased chance for conflicts.</li>
</ul>
<p>When you process data in batches, you keep the entire batch and any intermediate data in memory, which grows as you increase the batch size.
When you consume data with multiple flowlet instances which update the same values in datasets, there is a chance for a conflict.
The framework resolves them automatically, but in the end, you pay a price for the overhead caused by a retry.
As batch size grows, the chance for a conflict increases, as more rows are modified by overlapping transactions.
It is important to correctly partition data between flowlet instances to reduce the chance of conflict or to avoid it altogether.</p>
<p>One possible strategy is round-robin, which basically means that data is partitioned randomly and different
batches processed in parallel may have the same values and may result in updating the same cells in a dataset.
Using hash-partitioning instead of round-robin will help to resolve this.
Let’s take a look at the data flow of our application to see how we can employ a hash-partitioning strategy to avoid conflicts.</p>
<p>Suppose the Detector reads two slow disk reads for disk1.  It outputs “disk1” and again outputs “disk1”.
Since the default partitioning strategy is round-robin, Tracker1 takes the first “disk1” and Tracker2 takes the second “disk1”.
Since both Trackers are running simultaneously, they both read that “disk1” was slow 0 times, they both add one to that count of 0,
then both attempt to write what they think is the new value of 1. This is called a write conflict.
CDAP detects the conflict, allows only one write to go through, then replays the entire second event.
For example, CDAP may decide to let Tracker1 go through, which updates the slow count of disk1 to 1.
When Tracker2 tries to write, CDAP will detect the conflict, then replay the event.
Tracker2 reads “disk1” as input, gets the slow count of disk1 which has now been updated to 1, adds 1 to the count, and successfully writes the new value of 2.</p>
<p>Now pretend that we are using batches of 1000 instead of batches of 1.
Tracker1 takes a batch of 1000 and Tracker2 takes a separate batch of 1000.
The chance that Tracker1 has a disk in its batch that also appears in Tracker2’s batch is pretty high.
This means that when they both go to update their counts, only one of their updates will go through,
with the other needing to be replayed.
This means the work that one Tracker did will be entirely wasted and retried again,
which is much more costly with a big batch size because everything in the batch must be replayed.</p>
<p>One way to solve this problem is to make sure that no disks that go to Tracker1 ever go to Tracker2.
For example, all events for disk1 should go only to Tracker1, and never should go to Tracker2.
This is done by using hash partitioning instead of round-robin.
This is easy in CDAP and can be done in two lines.
When emitting in the Detector, a partition ID and key must be given in addition to the data being emitted.</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">DetectorFlowlet</span> <span class="kd">extends</span> <span class="n">AbstractFlowlet</span> <span class="p">{</span>
  <span class="p">...</span>

  <span class="nd">@ProcessInput</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">process</span><span class="p">(</span><span class="n">StreamEvent</span> <span class="n">diskMetrics</span><span class="p">)</span> <span class="p">{</span>
    <span class="p">...</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">readTime</span> <span class="o">&gt;</span> <span class="n">SLOW_THRESHOLD</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">out</span><span class="p">.</span><span class="na">emit</span><span class="p">(</span><span class="n">diskId</span><span class="p">,</span> <span class="s">&quot;diskId&quot;</span><span class="p">,</span> <span class="n">diskId</span><span class="p">);</span>
    <span class="p">}</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p>In the Tracker, you simply add the HashPartition annotation with the partition ID.</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">TrackerFlowlet</span> <span class="kd">extends</span> <span class="n">AbstractFlowlet</span> <span class="p">{</span>
  <span class="p">...</span>

  <span class="nd">@ProcessInput</span>
  <span class="nd">@Batch</span><span class="p">(</span><span class="mi">10</span><span class="p">)</span>
  <span class="nd">@HashPartition</span><span class="p">(</span><span class="s">&quot;diskId&quot;</span><span class="p">)</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">process</span><span class="p">(</span><span class="n">String</span> <span class="n">diskId</span><span class="p">)</span> <span class="p">{</span>
    <span class="p">...</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p>Now we can enjoy the benefits of larger batch sizes without worrying about wasted work due to write conflicts.
With batching and hash partitioning, our Detector and Tracker classes have changed just three lines with their final versions below:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">DetectorFlowlet</span> <span class="kd">extends</span> <span class="n">AbstractFlowlet</span> <span class="p">{</span>
  <span class="kd">private</span> <span class="kd">static</span> <span class="kd">final</span> <span class="kt">long</span> <span class="n">SLOW_THRESHOLD</span> <span class="o">=</span> <span class="mi">1000</span><span class="p">;</span>
  <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">NAME</span> <span class="o">=</span> <span class="s">&quot;slowReadDetector&quot;</span><span class="p">;</span>

  <span class="kd">private</span> <span class="n">OutputEmitter</span><span class="o">&lt;</span><span class="n">String</span><span class="o">&gt;</span> <span class="n">out</span><span class="p">;</span>

  <span class="nd">@ProcessInput</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">process</span><span class="p">(</span><span class="n">StreamEvent</span> <span class="n">diskMetrics</span><span class="p">)</span> <span class="p">{</span>
    <span class="n">String</span> <span class="n">event</span> <span class="o">=</span> <span class="n">Charsets</span><span class="p">.</span><span class="na">UTF_8</span><span class="p">.</span><span class="na">decode</span><span class="p">(</span><span class="n">diskMetrics</span><span class="p">.</span><span class="na">getBody</span><span class="p">()).</span><span class="na">toString</span><span class="p">();</span>
    <span class="c1">// events are expected to have the following format:</span>
    <span class="c1">// diskId operationTime (in microseconds)</span>
    <span class="n">String</span><span class="o">[]</span> <span class="n">fields</span> <span class="o">=</span> <span class="n">event</span><span class="p">.</span><span class="na">split</span><span class="p">(</span><span class="s">&quot; &quot;</span><span class="p">,</span> <span class="mi">2</span><span class="p">);</span>
    <span class="n">String</span> <span class="n">diskId</span> <span class="o">=</span> <span class="n">fields</span><span class="o">[</span><span class="mi">0</span><span class="o">]</span><span class="p">;</span>
    <span class="kt">long</span> <span class="n">readTime</span> <span class="o">=</span> <span class="n">Long</span><span class="p">.</span><span class="na">parseLong</span><span class="p">(</span><span class="n">fields</span><span class="o">[</span><span class="mi">1</span><span class="o">]</span><span class="p">);</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">readTime</span> <span class="o">&gt;</span> <span class="n">SLOW_THRESHOLD</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">out</span><span class="p">.</span><span class="na">emit</span><span class="p">(</span><span class="n">diskId</span><span class="p">,</span> <span class="s">&quot;diskId&quot;</span><span class="p">,</span> <span class="n">diskId</span><span class="p">);</span>
    <span class="p">}</span>
  <span class="p">}</span>
<span class="p">}</span>

<span class="kd">public</span> <span class="kd">class</span> <span class="nc">TrackerFlowlet</span> <span class="kd">extends</span> <span class="n">AbstractFlowlet</span> <span class="p">{</span>
  <span class="c1">// intentionally set very low for illustrative purposes</span>
  <span class="kd">private</span> <span class="kd">static</span> <span class="kd">final</span> <span class="kt">long</span> <span class="n">FLAG_THRESHOLD</span> <span class="o">=</span> <span class="mi">3</span><span class="p">;</span>
  <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">NAME</span> <span class="o">=</span> <span class="s">&quot;slowDiskTracker&quot;</span><span class="p">;</span>

  <span class="nd">@UseDataSet</span><span class="p">(</span><span class="s">&quot;slowDiskReads&quot;</span><span class="p">)</span>
  <span class="kd">private</span> <span class="n">KeyValueTable</span> <span class="n">slowDiskReadsTable</span><span class="p">;</span>

  <span class="nd">@UseDataSet</span><span class="p">(</span><span class="s">&quot;slowDisks&quot;</span><span class="p">)</span>
  <span class="kd">private</span> <span class="n">KeyValueTable</span> <span class="n">slowDisksTable</span><span class="p">;</span>

  <span class="nd">@ProcessInput</span>
  <span class="nd">@Batch</span><span class="p">(</span><span class="mi">100</span><span class="p">)</span>
  <span class="nd">@HashPartition</span><span class="p">(</span><span class="s">&quot;diskId&quot;</span><span class="p">)</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">process</span><span class="p">(</span><span class="n">String</span> <span class="n">diskId</span><span class="p">)</span> <span class="p">{</span>
    <span class="kt">byte</span><span class="o">[]</span> <span class="n">countAsBytes</span> <span class="o">=</span> <span class="n">slowDiskReadsTable</span><span class="p">.</span><span class="na">read</span><span class="p">(</span><span class="n">diskId</span><span class="p">);</span>
    <span class="kt">long</span> <span class="n">slowCount</span> <span class="o">=</span> <span class="n">countAsBytes</span> <span class="o">==</span> <span class="kc">null</span> <span class="o">?</span> <span class="mi">0</span> <span class="p">:</span> <span class="n">Bytes</span><span class="p">.</span><span class="na">toLong</span><span class="p">(</span><span class="n">countAsBytes</span><span class="p">);</span>
    <span class="n">slowCount</span><span class="o">++</span><span class="p">;</span>
    <span class="n">slowDiskReadsTable</span><span class="p">.</span><span class="na">write</span><span class="p">(</span><span class="n">diskId</span><span class="p">,</span> <span class="n">Bytes</span><span class="p">.</span><span class="na">toBytes</span><span class="p">(</span><span class="n">slowCount</span><span class="p">));</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">slowCount</span> <span class="o">==</span> <span class="n">FLAG_THRESHOLD</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">slowDisksTable</span><span class="p">.</span><span class="na">write</span><span class="p">(</span><span class="n">diskId</span><span class="p">,</span> <span class="n">Bytes</span><span class="p">.</span><span class="na">toBytes</span><span class="p">(</span><span class="n">System</span><span class="p">.</span><span class="na">currentTimeMillis</span><span class="p">()));</span>
    <span class="p">}</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
</div>
<div class="section" id="build-and-run-application">
<h3>Build and Run Application<a class="headerlink" href="#build-and-run-application" title="Permalink to this headline">🔗</a></h3>
<p>The DiskPerformanceApp can be built and packaged using the Apache Maven command:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> mvn clean package
</pre></div>
</div>
<p>Note that the remaining commands assume that the <code class="docutils literal notranslate"><span class="pre">cdap-cli.sh</span></code> script is available on your PATH.
If this is not the case, please add it:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> <span class="nb">export</span> <span class="nv">PATH</span><span class="o">=</span><span class="nv">$PATH</span>:&lt;CDAP home&gt;/bin
</pre></div>
</div>
<p>If you haven’t already started a standalone CDAP installation, start it with the command:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap.sh start
</pre></div>
</div>
<p>We can then deploy the application:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap-cli.sh load artifact target/cdap-flow-guide-&lt;version&gt;.jar
<span class="gp">$</span> cdap-cli.sh create app DiskPerformanceApp cdap-flow-guide &lt;version&gt; user
</pre></div>
</div>
<p>Next we start the flow:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap-cli.sh start flow DiskPerformanceApp.DiskPerformanceFlow
</pre></div>
</div>
<p>Note that there is one instance of the <code class="docutils literal notranslate"><span class="pre">Detector</span></code> Flowlet running and two instances of the <code class="docutils literal notranslate"><span class="pre">Tracker</span></code> Flowlet running:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap-cli.sh get flowlet instances DiskPerformanceApp.DiskPerformanceFlow.slowReadDetector
<span class="go">1</span>

<span class="gp">$</span> cdap-cli.sh get flowlet instances DiskPerformanceApp.DiskPerformanceFlow.slowDiskTracker
<span class="go">2</span>
</pre></div>
</div>
<p>We can scale out our application and increase the number of <code class="docutils literal notranslate"><span class="pre">Tracker</span></code> Flowlets to four:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap-cli.sh <span class="nb">set</span> flowlet instances DiskPerformanceApp.DiskPerformanceFlow.slowDiskTracker <span class="m">4</span>
<span class="go">Successfully set flowlet &#39;DiskPerformanceFlow&#39; of flow &#39;slowDiskTracker&#39; of app &#39;DiskPerformanceApp&#39; to 4 instances</span>

<span class="gp">$</span> cdap-cli.sh get flowlet instances DiskPerformanceApp.DiskPerformanceFlow.slowDiskTracker
<span class="go">4</span>
</pre></div>
</div>
<p>Scaling your application is easy in CDAP!
Now we can manually send enough slow disk events to the diskReads stream to get a disk classified as a slow disk:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap-cli.sh send stream diskReads <span class="se">\&#39;</span><span class="s1">&#39;disk1 1001&#39;</span><span class="se">\&#39;</span>
<span class="gp">$</span> cdap-cli.sh send stream diskReads <span class="se">\&#39;</span><span class="s1">&#39;disk1 1001&#39;</span><span class="se">\&#39;</span>
<span class="gp">$</span> cdap-cli.sh send stream diskReads <span class="se">\&#39;</span><span class="s1">&#39;disk1 1001&#39;</span><span class="se">\&#39;</span>
<span class="gp">$</span> cdap-cli.sh send stream diskReads <span class="se">\&#39;</span><span class="s1">&#39;disk1 1001&#39;</span><span class="se">\&#39;</span>
<span class="gp">$</span> cdap-cli.sh send stream diskReads <span class="se">\&#39;</span><span class="s1">&#39;disk1 1001&#39;</span><span class="se">\&#39;</span>
</pre></div>
</div>
<p>Next we start the service:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap-cli.sh start service DiskPerformanceApp.DiskPerformanceService
</pre></div>
</div>
<p>The Service exposes a RESTful API that allows us to display all slow disks and the timestamp at which they were flagged as a slow disk.
Make the request to query slow disks:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> curl -w<span class="s1">&#39;\n&#39;</span> http://localhost:10000/v3/namespaces/default/apps/DiskPerformanceApp/services/DiskPerformanceService/methods/slowdisks
</pre></div>
</div>
<p>Example output:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="go">{&quot;disk1&quot;:&quot;2015-04-04 13:46:33 PDT&quot;}</span>
</pre></div>
</div>
</div>
</div>
<div class="section" id="extend-this-example">
<h2>Extend This Example<a class="headerlink" href="#extend-this-example" title="Permalink to this headline">🔗</a></h2>
<p>To make this application more useful, you can extend it by:</p>
<ul class="simple">
<li>Including the disk type in the Stream event and categorize a slow read based on the type of disk.</li>
<li>Passing your own custom Java object through the Flowlets instead of a String.</li>
<li>Adding an endpoint to the Service that can remove a disk from the <code class="docutils literal notranslate"><span class="pre">slowDisks</span></code> Dataset.</li>
<li>Changing the logic so that 1000 normal disk read times counteract a slow disk read.</li>
<li>Tracking additional disk metrics, such as write times, and use a combination of factors
to determine whether or not a disk belongs in the <code class="docutils literal notranslate"><span class="pre">slowDisks</span></code> table.</li>
</ul>
</div>
<div class="section" id="share-and-discuss">
<h2>Share and Discuss!<a class="headerlink" href="#share-and-discuss" title="Permalink to this headline">🔗</a></h2>
<p>Have a question? Discuss at the <a class="reference external" href="https://groups.google.com/forum/#!forum/cdap-user">CDAP User Mailing List</a>.</p>
</div>
<div class="section" id="license">
<h2>License<a class="headerlink" href="#license" title="Permalink to this headline">🔗</a></h2>
<p>Copyright © 2014-2015 Cask Data, Inc.</p>
<p>Licensed under the Apache License, Version 2.0 (the “License”); you may
not use this file except in compliance with the License. You may obtain
a copy of the License at</p>
<p><a class="reference external" href="http://www.apache.org/licenses/LICENSE-2.0">http://www.apache.org/licenses/LICENSE-2.0</a></p>
<p>Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an “AS IS” BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.</p>
</div>
</div>

</div>
    <div class="col-md-2">
      <div id="right-sidebar" class="bs-sidenav scrollable-y" role="complementary">
        <div id="localtoc-scrollspy">
        </div>
      </div>
    </div></div>
</div>
<!-- block main content end -->
<!-- block footer -->
<footer class="footer">
      <div class="container">
        <div class="row">
          <div class="col-md-2 footer-left"><a title="Creating Hydrator Applications using CDAP System Artifacts" href="cdap-etl-guide.html" />Previous</a></div>
          <div class="col-md-8 footer-center"><a class="footer-tab-link" href="../table-of-contents/../../reference-manual/licenses/index.html">Copyright</a> &copy; 2014-2020 Cask Data, Inc.&bull; <a class="footer-tab-link" href="//docs.cask.co/cdap/6.1.1/cdap-docs-6.1.1-web.zip" rel="nofollow">Download</a> an archive or
<a class="footer-tab-link" href="//docs.cask.co/cdap">switch the version</a> of the documentation
          </div>
          <div class="col-md-2 footer-right"><a title="Ingesting Data into CDAP using Apache Flume" href="cdap-flume-guide.html" />Next</a></div>
        </div>
      </div>
    </footer>
<!-- block footer end -->
<script type="text/javascript" src="../_static/bootstrap-3.3.6/js/bootstrap.min.js"></script><script type="text/javascript" src="../_static/js/bootstrap-sphinx.js"></script><script type="text/javascript" src="../_static/js/abixTreeList-2.js"></script><script type="text/javascript" src="../_static/js/cdap-dynamicscrollspy-4.js"></script><script type="text/javascript" src="../_static/js/cdap-version-menu.js"></script><script type="text/javascript" src="../_static/js/copy-to-clipboard.js"></script><script type="text/javascript" src="../_static/js/jquery.mousewheel.min.js"></script><script type="text/javascript" src="../_static/js/jquery.mCustomScrollbar.js"></script><script type="text/javascript" src="../_static/js/js.cookie.js"></script><script type="text/javascript" src="../_static/js/tabbed-parsed-literal-0.2.js"></script><script type="text/javascript" src="../_static/js/cdap-onload-javascript.js"></script><script type="text/javascript" src="../_static/js/cdap-version-menu.js"></script>
    <script src="https://cdap.gitee.io/docs/cdap/json-versions.js"/></script>
  </body>
</html>